package com.camemax.kafkatest;

import ch.qos.logback.classic.LoggerContext;
import com.camemax.pojo.ItemViewCount;
import com.camemax.pojo.UserBehavior;
import com.camemax.utils.HotItemOperators;
import com.camemax.utils.StreamEnvUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.logging.log4j.Level;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;

public class KafkaConsumerHotItems {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment streamEnv = StreamEnvUtils.getStreamEnv(1);

        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "172.28.40.190:9092");
        props.setProperty("group.id","test-group");
        props.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("auto.offset.reset","latest");

        DataStreamSource<String> kafkaDataStream = streamEnv.addSource(new FlinkKafkaConsumer<>(
                "hotItems",
                new SimpleStringSchema(),
                props
        ));
        // 添加时间戳分配器
        DataStream<UserBehavior> watermarkDataStream = kafkaDataStream.map(line -> {
            String[] fields = line.split(",");
            return new UserBehavior(Long.valueOf(fields[0]), Long.valueOf(fields[1]), Integer.valueOf(fields[2]), fields[3], Long.valueOf(fields[4]));
        }).assignTimestampsAndWatermarks(
                WatermarkStrategy.<UserBehavior>noWatermarks()
                        .withTimestampAssigner((SerializableTimestampAssigner<UserBehavior>) (element, recordTimestamp) -> element.getTimestamp() * 1000L)
        );

        // 分组聚合开窗
        DataStream<ItemViewCount> windowAggregateStream = watermarkDataStream.filter(data -> "pv".equals(data.getBehavior()))
                .keyBy(UserBehavior::getItemId)
                .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(5))) // 滑动窗口实现以每小时分桶、每五分钟滑动
                // aggregate方法Prefix部分 —— 聚合函数
                // aggregate方法Suffix部分 —— 窗口函数
                .aggregate(new HotItemOperators.UserBehaviorAggregateFunctionImpl(),
                        new HotItemOperators.ItemViewCountWindowFunctionImpl());

        SingleOutputStreamOperator<String> resultDataStream = windowAggregateStream.keyBy(ItemViewCount::getTimstamp)
                .process(new HotItemOperators.ItemViewCountKeyedProcessFunctionImpl(5));

        resultDataStream.print();

        streamEnv.execute();
    }
}
